/**
 * Licensed under:
 * 
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 * 
 * @author Emanuele Gherardini (e.gherardini@yahoo.it / e.gherardini@hotmail.com)
 */
package org.apache.camel.component.coherence.topic;

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;

import org.apache.camel.Exchange;
import org.apache.camel.Processor;
import org.apache.camel.impl.DefaultConsumer;
import org.apache.camel.impl.DefaultExchangeHolder;
import org.apache.camel.impl.DefaultExecutorServiceStrategy;

import com.oracle.coherence.patterns.messaging.MessagingSession;
import com.oracle.coherence.patterns.messaging.Subscriber;
import com.oracle.coherence.patterns.messaging.SubscriptionIdentifier;
import com.oracle.coherence.patterns.messaging.TopicSubscriptionConfiguration;

public class CoherenceTopicConsumer extends DefaultConsumer {
	
	/**
	 * Maximum time this subscriber will wait before
	 * forcing termination.
	 */
	public static final long TERMINATION_WAIT_TIME_IN_MILLIS = 5000l;
	
	private final CoherenceTopicEndpoint topicEndpoint;
	private final TopicSubscriptionConfiguration subscriptionConfiguration;
	
	/**
	 * Since the call to get a message from a CoherenceQueue is blocking,
	 * the effective subscriber is started in a separate thread.
	 * This executor is used to handle the subscriber thread.
	 */
	private ExecutorService executor;
	
private final int concurrentConsumers;
	
	/**
	 * This map holds all the subscribers to the queue.
	 * It is used on shutdown to unsubscribe them all.
	 */
	private final ConcurrentMap<SubscriptionIdentifier, Subscriber> subscribersList;
	
	public CoherenceTopicConsumer(CoherenceTopicEndpoint endpoint, TopicSubscriptionConfiguration subscriptionConfiguration, Processor processor) {
		this(endpoint,subscriptionConfiguration,processor,1);
	}
	
	public CoherenceTopicConsumer(CoherenceTopicEndpoint endpoint, TopicSubscriptionConfiguration subscriptionConfiguration, Processor processor, int concurrentConsumers) {
		
		super(endpoint, processor);
		
		this.topicEndpoint = endpoint;
		this.subscriptionConfiguration = subscriptionConfiguration;
		this.executor = new DefaultExecutorServiceStrategy(endpoint.getCamelContext()).newSingleThreadExecutor(this, "CoherenceQueueConsumerThread");
		this.concurrentConsumers = concurrentConsumers;
		this.subscribersList = new ConcurrentHashMap<SubscriptionIdentifier, Subscriber>(concurrentConsumers);
		
		this.executor = endpoint.getCamelContext().getExecutorServiceStrategy().newFixedThreadPool(this, "CoherenceTopicConsumerThread", concurrentConsumers);
		
		log.info("CoherenceQueueConsumer succesfully created: " + this);
	}

	
	@Override
	protected void doStart() throws Exception {
		
		super.doStart();
		
		startConsumers();
		
		if(log.isDebugEnabled())
			log.debug("["+this+"] started");
		
	}
	
	private void startConsumers() {
		CoherenceTopicConsumerThread task = new CoherenceTopicConsumerThread();
		for (int i = 0; i < concurrentConsumers; i++) {
			executor.submit(task);
		}
	}
	
	@Override
	protected void doStop() throws Exception {
		
		if(log.isDebugEnabled())
			log.debug("Terminating ["+this+"]...");
		
		unsubscribeConsumers();
		
		if(log.isDebugEnabled())
			log.debug("["+this+"] consumers succesfully unsubscribed, waiting for thread termination");
		
		this.topicEndpoint.getCamelContext().getExecutorServiceStrategy().shutdown(executor);
		
		boolean executorGracefullyTerminated = executor.awaitTermination(TERMINATION_WAIT_TIME_IN_MILLIS, TimeUnit.MILLISECONDS);
		
		//if the thread is stuck on the "getMessage()" call, force termination
		if(!executorGracefullyTerminated) {
			log.warn("Forcing termination of  ["+this+"]");
			this.topicEndpoint.getCamelContext().getExecutorServiceStrategy().shutdownNow(executor);
		}
		
		super.doStop();
		
		if(log.isDebugEnabled())
			log.debug("["+this+"] succesfully terminated");
		
	}
	
	private void unsubscribeConsumers() {
		for(SubscriptionIdentifier id : subscribersList.keySet()) {
			subscribersList.get(id).unsubscribe();
		}
	}
	
	
	private class CoherenceTopicConsumerThread extends Thread {
		
		@Override
		public void run() {
			Subscriber subscription = subscribe();
			consumeMessages(subscription);
		}
		
		private Subscriber subscribe() {
			MessagingSession session = topicEndpoint.getMessagingSession();
			Subscriber subscription = session.subscribe(topicEndpoint.getTopicIdentifier(), subscriptionConfiguration);
			subscription.setAutoCommit(false);
			
			subscribersList.put(subscription.getSubscriptionIdentifier(),subscription);
			return subscription;
		}
		
		private void consumeMessages(Subscriber subscription) {
			
			if(log.isDebugEnabled())
				log.debug("Consumer ["+this+"] is consuming messages...");
			
			while(isRunAllowed() && subscription.isActive()) {
				
				Object rawMessage = subscription.getMessage();
				
				log.info("Received Message: [" +rawMessage+"]");
				
				Exchange ex = getEndpoint().createExchange();
				
				try {
					DefaultExchangeHolder.unmarshal(ex, (DefaultExchangeHolder) rawMessage);
					getProcessor().process(ex);
					subscription.commit();
				} catch (Exception e) {//if this exception is a SubscriberInterruptedException, the message will be rolled back.
					subscription.rollback();
					getExceptionHandler().handleException(
                            "Exeception occurred consuming message. The message has been rolled back.", ex, 
                            ex.getException());
				}
				
			}
			
		}
		
	}
	
	
	@Override
	public String toString() {
		return "CoherenceTopicConsumer [topicEndpoint=" + topicEndpoint	+ "] + concurrentConsumers="+concurrentConsumers;
	}
	
	
}
